package com.xiaotu.spider.pipeline;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import com.xiaotu.common.model.*;
import org.apache.commons.lang.StringUtils;
import org.springframework.stereotype.Component;

import us.codecraft.webmagic.ResultItems;
import us.codecraft.webmagic.Task;

import com.xiaotu.common.util.Constants;
import com.xiaotu.common.util.DataRedisKeyUtils;
import com.xiaotu.common.util.DateUtils;
import com.xiaotu.common.util.KeyLocker;
import com.xiaotu.common.util.StringUtil;
import com.xiaotu.common.util.Constants.DataRedisKey;
import com.xiaotu.spider.APipeline;
import com.xiaotu.storage.converter.AConverter;
import com.xiaotu.storage.converter.ConverterFactory;
import com.xiaotu.storage.converter.DateConverter;

import net.sf.json.JSONObject;

/**
 * @类名 APipeLine
 * @日期 2017年2月10日
 * @作者 高海军
 * @功能 网站过长数据pipeline
 */
@Component("TVSitePipeline")
public class TVSitePipeline extends APipeline
{

    @Override
    protected void doProcess(ResultItems resultItems, Task task)
    {
        this.tvRecordProcess(resultItems);

        List<TVPubCommentModel> commentList = resultItems
                .get(TVPubCommentModel.class.getName());// 评论内容
        this.tvCommentProcess(commentList);

        List<ProfessionalCommentModel> professionalCommentModelList = resultItems
                .get(ProfessionalCommentModel.class.getName());// 专业影评
        this.tvProCommentProcess(professionalCommentModelList);

        this.tvSetProcess(resultItems);// 分集梗概

        this.tvPeopleProcess(resultItems);//人群受众

    }

    private void tvPeopleProcess(ResultItems resultItems)
    {
        TVAreaModel tvArea = resultItems.get(TVAreaModel.class.getName());
        if (tvArea != null)
        {
            String key = DataRedisKeyUtils.getKey(DataRedisKey.TV_AREA_YYYYMMDD);
            cacheHandler.hSet(key, tvArea.getTvId(), tvArea);
            cacheHandler.setCacheTimeout(key);
        }

        TVAudienceModel tvPeople = resultItems.get(TVAudienceModel.class.getName());
        if (tvPeople != null)
        {
            String key = DataRedisKeyUtils.getKey(DataRedisKey.TV_PEOPLE_YYYYMMDD);
            cacheHandler.hSet(key, tvPeople.getTvId(), tvPeople);
            cacheHandler.setCacheTimeout(key);
        }
    }

    private void tvRecordProcess(final ResultItems resultItems)
    {
        final TVRecordModel tvRecord = resultItems
                .get(TVRecordModel.class.getName());// 评分、播放量、赞、踩
        List<TVRecordModel> tvList = resultItems.get(TVRecordModel.class.getName() + "List");
        if (tvRecord == null && tvList == null)
            return;
        final List<TVRecordModel> list = new ArrayList<>();
        if (tvRecord != null)
            list.add(tvRecord);
        if (tvList != null)
            list.addAll(tvList);
        final String key = DataRedisKeyUtils
                .getKey(Constants.DataRedisKey.TV_NetPlay_YYYYMMDD);
        final String field = DataRedisKeyUtils.getTVRecordKey(list.get(0));

        KeyLocker.lock(field, new Runnable()
        {
            @Override
            public void run()
            {
                for (TVRecordModel record : list)
                {
                    Map<String, Object> recordMap = cacheHandler.hGet(key, field,
                            Map.class);
                    if (recordMap == null)
                        recordMap = new HashMap<String, Object>();
                    recordMap.put(record.getTvName(), record);

                    recordMap.put("tvId", record.getTvId());
                    Object name = resultItems.getRequest().getExtra("tvName");
                    recordMap.put("tvName",
                            name == null ? record.getTvName() : name.toString());

                    cacheHandler.hSet(key, field, recordMap);
                    cacheHandler.setCacheTimeout(key);
                }
            }
        });
    }

    /**
     * 评论内容存储
     *
     * @param commentList 评论内容列表
     */
    private void tvCommentProcess(List<TVPubCommentModel> commentList)
    {
        if (commentList == null || commentList.isEmpty())
            return;

        Map<String, TVPubCommentModel> commentMap = new HashMap<String, TVPubCommentModel>();
        Map<String, String> historyMap = new HashMap<String, String>();
        for (TVPubCommentModel pubComment : commentList)
        {
            if (this.validatePubComment(pubComment))
            {
                String field = DataRedisKeyUtils.getTVPubCommentKey(pubComment);
                commentMap.put(field, pubComment);
                historyMap.put(field, StringUtils.EMPTY);
            }

        }

        if (commentMap.isEmpty())
            return;

        String key = DataRedisKeyUtils
                .getKey(Constants.DataRedisKey.TV_Content_YYYYMMDD);

        cacheHandler.hMSet(key, commentMap);
        this.cacheHandler.hMSet(
                DataRedisKey.Public_Sentiment_History.toString(), historyMap);

        cacheHandler.setCacheTimeout(
                DataRedisKey.Public_Sentiment_History.toString());
        cacheHandler.setCacheTimeout(key);

    }

    /**
     * 专业影评存储
     *
     * @param proCommentList 影评列表
     */
    private void tvProCommentProcess(
            List<ProfessionalCommentModel> proCommentList)
    {
        if (proCommentList == null || proCommentList.isEmpty())
            return;

        Map<String, ProfessionalCommentModel> commentMap = new HashMap<String,
                ProfessionalCommentModel>();
        Map<String, String> historyMap = new HashMap<String, String>();
        for (ProfessionalCommentModel comment : proCommentList)
        {
            if (this.validateProComment(comment))
            {
                String field = DataRedisKeyUtils.getTVProCommentKey(comment);

                commentMap.put(field, comment);
                historyMap.put(field, StringUtils.EMPTY);
            }
        }

        if (commentMap.isEmpty())
            return;

        String key = DataRedisKeyUtils.getKey(
                Constants.DataRedisKey.TV_Professional_Content_YYYYMMDD);

        cacheHandler.hMSet(key, commentMap);
        this.cacheHandler.hMSet(
                DataRedisKey.TV_Professional_Content_History.toString(),
                historyMap);

        cacheHandler.setCacheTimeout(
                DataRedisKey.TV_Professional_Content_History.toString());
        cacheHandler.setCacheTimeout(key);

    }

    /**
     * 分集梗概存储
     *
     * @param resultItems 分集梗概
     */
    private void tvSetProcess(ResultItems resultItems)
    {
        String setKey = Constants.DataRedisKey.TV_SetsPlot.toString();
        TVSetsPlotModel tvSet = resultItems
                .get(TVSetsPlotModel.class.getName());// 一个分集梗概
        if (tvSet != null)
        {
            cacheHandler.hSet(setKey, DataRedisKeyUtils.getTVSetPlotsKey(tvSet),
                    tvSet);
            cacheHandler.setCacheTimeout(setKey);
        }

        List<TVSetsPlotModel> setsList = resultItems
                .get(TVSetsPlotModel.class.getName() + "List");// 多个分集梗概
        if (setsList == null || setsList.isEmpty())
            return;

        Map<String, TVSetsPlotModel> setMap = new HashMap<String, TVSetsPlotModel>();
        for (TVSetsPlotModel set : setsList)
            setMap.put(DataRedisKeyUtils.getTVSetPlotsKey(set), set);

        cacheHandler.hMSet(setKey, setMap);
        cacheHandler.setCacheTimeout(setKey);
    }

    /**
     * 校验评论是否为有效数据
     *
     * @param comment 评论对象
     * @return 是否
     */
    private boolean validatePubComment(TVPubCommentModel comment)
    {
        if (!StringUtil.allNotEmpty(comment.getResourceSiteId(),
                comment.getTvId(), comment.getCommentDate(),
                comment.getCommentUser(), comment.getContent()))
            return false;
        String commentDate = (String) ConverterFactory.convert(
                JSONObject.fromObject(comment), "commentDate",
                AConverter.DATE_CONVERT, true);
        if (StringUtils.isEmpty(commentDate))
            return false;

        if (DateUtils
                .fomartStringToDate("2015-08-01 00:00:00",
                        DateConverter.DATE_FORMATE)
                .after(DateUtils.fomartStringToDate(commentDate,
                        DateConverter.DATE_FORMATE)))
            return false;

        comment.setCommentDate(commentDate);
        String key = DataRedisKeyUtils.getTVPubCommentKey(comment);// 获取唯一标识以判断是否已存在
        if (key == null)
            return false;
        return !cacheHandler
                .hExists(DataRedisKey.Public_Sentiment_History.toString(), key);

    }

    /**
     * 校验影评是否为有效数据
     *
     * @param comment 评论对象
     * @return 是否
     */
    private boolean validateProComment(ProfessionalCommentModel comment)
    {
        if (!StringUtil.allNotEmpty(comment.getResourcesiteid(),
                comment.getTvId(), comment.getCommentdate(),
                comment.getCommentuser(), comment.getTitle()))
            return false;
        String commentDate = (String) ConverterFactory.convert(
                JSONObject.fromObject(comment), "commentdate",
                AConverter.DATE_CONVERT, true);
        if (StringUtils.isEmpty(commentDate))
            return false;

        comment.setCommentdate(commentDate);
        String key = DataRedisKeyUtils.getTVProCommentKey(comment);// 获取唯一标识以判断是否已存在
        if (key == null)
            return false;
        return !cacheHandler.hExists(
                DataRedisKey.TV_Professional_Content_History.toString(), key);

    }
}